package com.niit.service

import java.text.SimpleDateFormat

import com.niit.bean.AdClickData
import com.niit.dao.BlackListDao
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.DStream

/*
需求：将每天对某个广告点击超过100次的用户拉黑，将黑名单数据保存到MySQL当中。

1.读取Kafka数据之后，并对MySQL中存储黑名单数据做校验；
2.校验通过则对用户点击广告次数累加一并存入当MySQL当中
3.在存入MySQL之后数据校验，如果单日超过100次则将该用户拉入黑名单
 */
class BlackListService {



  // Service层专注于数据分析，所以处理获取数据的代码不应该写在这里
  def dataAnalysis(data: DStream[AdClickData]): Any = {
    // 1714967973680  华南  深圳  4   3
    val kdata: DStream[((String, String, String), Int)] = filterKafkaData(data)
    checkUserCount(kdata)
  }
  /*
  利用在Kafka中采集出来流式数据对数据库进行实时检索
  将已经在黑名单中的数据过滤掉，和统计每个用户在同一天点击同一个广告的总次数
   */
  def filterKafkaData(ds: DStream[AdClickData]): DStream[ ( (String,String,String), Int  )  ] = {
    ds.transform(rdd=>{
        //1.判断用户是否已经在黑名单当中，如果该用户在黑名单中，就过滤掉
      val filterRdd: RDD[AdClickData] = rdd.filter(data => {
        //在数据库中查询该用户Id是否存在黑名单表
        val blackListDao = new BlackListDao
        !blackListDao.selectBlackUserById(data.user)

      })
      //filterRdd 不在黑名单中的用户数据
       // 1714967973680  华南  深圳  4   3  转换成 每个用户在同一天点击同一个广告的 一次
      val reduceData: RDD[((String, String, String), Int)] = filterRdd.map(data => {
        val user = data.user //张三
        val ad = data.ad  // 加多宝
        //要对时间戳进行格式转换  2024-05-04
        val sdf = new SimpleDateFormat("yyyy-MM-dd")
        val day = sdf.format(new java.util.Date(data.ts.toLong)) // 2024-05-04 spark Context Shutdown
         //    ( (2024-05-04,张三,加多宝) , 1)
        ((day, user, ad), 1)
      })
      //( (2024-05-04,张三,加多宝) , 1)  ( (2024-05-04,张三,加多宝) , 1)  ( (2024-05-04,张三,加多宝) , 1)
      reduceData.reduceByKey(_ + _) //( (2024-05-04,张三,加多宝) , 3) 可以到某个用户当天点击某个广告总次数
    })
  }

  /*
      检查用户的点击次数
      //如果 不满足拉入黑名单的条件存入 每日用户点击广告次数的数据表中，当存入每日用户点击广告次数的数据表发现满足拉黑的条件也将该用户拉入黑名单
       在上一步统计广告总次数已经满足拉黑的次数，就直接拉黑
   */
  private def checkUserCount(ds: DStream[( (String,String,String), Int  )    ]):Unit={
    //DStream 必须在最后一步要有输出， print saveAsXxxx foreachRDD
    ds.foreachRDD(rdd=>{
      rdd.foreach{
        case ( (day,user,ad),count)=>{
          println(s"${day},${user},${ad},${count}")
        val blackListDao = new BlackListDao
          //如果点击次数大于等于100 则拉黑
          if(count >= 100){
            //将该user 存入黑名单数据表中
            blackListDao.insertBlackList(user)
          }else{
            //存入每日用户点击广告次数
            blackListDao.insertUserAdCount(day,user,ad ,count)
          }
          //判断更新后的点击数据是否超过100次，如果也要拉黑。
          //查询在某天某用户某广告 点击超过100次 ，也要拉黑
          val bool: Boolean = blackListDao.checkUserCount(day, user, ad, 100)
          if(bool){
            blackListDao.insertBlackList(user)
          }

        }
      }
    })


  }

}
